package com.example.mq.mqserver.core;

import com.example.mq.common.Consumer;
import com.example.mq.common.ConsumerEnv;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.VirtualHost;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: DongGuoZhen
 * @Date: 2024/03/06/14:12
 * @Description: 这个类来实现消费消息的核心逻辑
 */
public class ConsumerManager {
    //持有一个上层的virtualHost 用来操作数据
    private VirtualHost parent;

    //持有一个线程池,执行具体的回调任务
    private ExecutorService workerPool = Executors.newFixedThreadPool(4);

    //存放令牌的队列
    private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();

    //扫描线程
    private Thread scannerThread = null;

    public ConsumerManager(VirtualHost p) {
        parent = p;
        scannerThread = new Thread(()->{
            while (true) {
                try {
                    //拿到令牌对象
                    String queueName = tokenQueue.take();
                    //根据令牌找到队里
                    MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
                    if(queue == null) {
                        throw new MqException("[ConsumerManager] 取令牌后发现该队列名不存在! queueName=" + queueName);
                    }
                    //从这个队列中消费一个消息
                    synchronized (queue) {
                        consumeMessage(queue);
                    }
                } catch (InterruptedException | MqException e) {
                    e.printStackTrace();
                }
            }
        });
        //把这个线程设置为后台线程  这个线程并不会响应到前台线程
        scannerThread.setDaemon(true);
        scannerThread.start();
    }

    //这个方法的调用时机就是发送消息的时候
    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }


    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        //找到对应的队列
        MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
        if (queue == null) {
            throw new MqException("[ConsumerManager] 队列不存在! queueName = " + queueName);
        }
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag, queueName, autoAck, consumer);
        queue.addConsumerEnv(consumerEnv);
        //如果当前队列中已经有了一些消息了,需要立即消费掉
        int n = parent.getMemoryDataCenter().getMessageCount(queueName);
        for (int i = 0; i < n; i++) {
            consumeMessage(queue);
        }
    }

    private void consumeMessage(MSGQueue queue) {
        //1. 先按照轮询的方式,找个消费者出来
        ConsumerEnv luckyDog = queue.choseConsumer();
        if(luckyDog == null) {
            //当前队列没有订阅者 暂时不消费,等后面有消费者出现
            return;
        }
        //2. 从队列中取出一个消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName()) ;
        if(message == null) {
            //当前队列中没有消息 也不要消费
            return;
        }
        //3. 把这个消息放到消费者的回调方法中,丢给线程池取执行
        workerPool.submit(()->{
            try{
                //1. 先把消息放在待确认的集合中 这个操作必须要在执行回调之前   考虑消息丢失问题
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(), message);
                //2. 真正的执行回调
                luckyDog.getConsumer().handleDelivery(luckyDog.getConsumerTag(),message.getBasicProperties(),message.getBody());
                //3. 如果当前是自动应答,此时就可以直接删除消息,如果是手动应答,此处暂时不处理.交给后续的消费者调用basicAck方法来处理
                if(luckyDog.isAutoAck()) {
                    // 删除硬盘上的消息  判定是否为持久化  2 持久化   1 不持久化
                    if(message.getDeliverMode() ==2) {
                        //删除硬盘上的消息
                        parent.getDiskDataCenter().deleteMessage(queue,message);
                    }
                    //删除上面的待确认集合中的消息
                    parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(), message.getMessageId());
                    //删除内存中消息中心的消息   key是messageId,value是message对象
                    //private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
                    parent.getMemoryDataCenter().removeMessage(message.getMessageId());
                    System.out.println("[ConsumerManager] 消息被成功消费! queueName=" + queue.getName());
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }
}
